package com.lzq.mq.broker.model;

import com.lzq.mq.broker.cache.CommonCache;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.HashMap;
import java.util.Map;

import static com.lzq.mq.common.constants.BrokerConstants.CONSUME_QUEUE_ELEMENT_BYTE_SIZE;

/**
 * 存储消费队列的偏移量
 */
@Data
public class ConsumeQueueOffsetModel {

    /**
     * 存储主题下各个消费组的各个成员队列的偏移量
     */
    private OffsetTable offsetTable = new OffsetTable();

    /**
     * 存储主题下各个消费组的各个成员队列的偏移量
     * 成员: topicConsumerGroupDetailMap (Map<String, Map<String, Map<String, String>>>)
     * key: topicName
     * value: topic下各个消费组的队列偏移量map
     */
    @Data
    public static class OffsetTable {
        Map<String, ConsumerGroupDetail> topicConsumerGroupDetailMap;
    }

    /**
     * 主题下各个消费组的偏偏移量map
     * 成员: consumerGroupDetailMap (Map<String, Map<String, String>>)
     * key: topicName
     * value: topic下各个消费组的队列偏移量map
     */
    @Data
    public static class ConsumerGroupDetail {

        /**
         * 存储消费组的队列偏移量
         * key: 消费组名称
         * value: 消费组中对各个消费队列的偏移量map
         * 偏移量格式: 00000000#9792 : 文件id#偏移量
         */
        private Map<String, Map<String, String>> consumerGroupDetailMap;
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class ConsumeQueueOffsetInfo {
        private Integer fileName;
        private Integer offset;
    }

    /**
     * 获取主题下的消费者组能够消费某个队列的偏移量
     * @param topicName 主题名称
     * @param consumeGroup 消费组名称
     * @param queueId 队列ID
     * @return 队列的偏移量
     */
    public String getConsumeQueueOffsetInfoStr(String topicName, String consumeGroup, Integer queueId) {
        return offsetTable.getTopicConsumerGroupDetailMap()
                .get(topicName).getConsumerGroupDetailMap()
                .get(consumeGroup).get(String.valueOf(queueId));
    }

    /**
     * 获取主题下的消费者组能够消费某个队列的偏移量
     * @param topicName 主题名称
     * @param consumeGroup 消费组名称
     * @param queueId 队列ID
     * @return 队列的偏移量
     */
    public ConsumeQueueOffsetInfo getConsumeQueueOffsetInfo(String topicName, String consumeGroup, Integer queueId) {

        try {
            String infoStr = getConsumeQueueOffsetInfoStr(topicName, consumeGroup, queueId);
            String[] split = infoStr.split("#");
            return new ConsumeQueueOffsetInfo(Integer.valueOf(split[0]), Integer.valueOf(split[1]));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取主题下的消费者组能够消费某个队列的最多元素个数
     * @param topicName 主题名称
     * @param consumeGroup 消费组
     * @param queueId 队列ID
     * @return 元素个数
     */
    public int getConsumeAvailableElementNum(String topicName, String consumeGroup, Integer queueId) {
        ConsumeQueueModel consumeQueueModel = CommonCache.getMqTopicModelMap().get(topicName).getQueueList().get(queueId);
        ConsumeQueueOffsetInfo offsetInfo = getConsumeQueueOffsetInfo(topicName, consumeGroup, queueId);
        return (consumeQueueModel.getLatestOffset().get() - offsetInfo.getOffset()) / CONSUME_QUEUE_ELEMENT_BYTE_SIZE;
    }

}
